package com.zmn.oms.task.kafka.process;

import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.api.config.annotation.NacosValue;
import com.google.common.collect.Maps;
import com.zmn.common.utils.number.NumberUtil;
import com.zmn.common.utils.string.StringUtil;
import com.zmn.manager.mq.constant.MqTagConsts;
import com.zmn.manager.mq.constant.MqTopicConsts;
import com.zmn.manager.mq.interfaces.ZmnMQSender;
import com.zmn.oms.common.constant.OrderConsts;
import com.zmn.oms.constant.MessageQueueTopicConstant;
import com.zmn.oms.manager.utils.OmsMqTagConsts;
import com.zmn.oms.task.kafka.avro.Field;
import com.zmn.oms.task.kafka.avro.Integer;
import com.zmn.oms.task.kafka.avro.Operation;
import com.zmn.oms.task.kafka.avro.Record;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;

/**
 * 类描述：dts 数据消费
 *
 * @author lujia
 * @date 2019/8/15 20:22
 */
@Slf4j
@Service
public class RecordProcessServiceImpl implements RecordProcessService {

    /**
     * 配件表
     */
    private static String PART_CHANGE_TABLE = "zmn_oms.oms_order_part";
    /**
     * 评价扩展表
     */
    private static String COMMENT_CHANGE_TABLE = "zmn_oms.oms_order_comment_extend";
    /**
     * 工单表
     */
    private static String WORK_CHANGE_TABLE = "zmn_oms.oms_order_work";

    /**
     * 工单支付凭证表
     */
    private static String PAY_CHANGE_TABLE = "zmn_oms.oms_order_pay";

    private static Map<String, String> tableMap = Maps.newHashMap();

    static {

        // 这里维护订阅表与分片id映射
//        tableMap.put("zmn_oms.oms_order", "order_id");
        tableMap.put("zmn_oms.oms_order_work", "order_id");
        tableMap.put("zmn_oms.oms_order_detail", "detail_id");
        tableMap.put("zmn_oms.oms_order_member", "member_order_id");
        tableMap.put("zmn_oms.oms_order_work_master", "order_id");
        tableMap.put("zmn_oms.oms_order_product", "order_id");
        tableMap.put("zmn_oms.oms_order_service_item", "order_id");
        //tableMap.put("zmn_oms.oms_order_discount", "order_id");
        tableMap.put("zmn_oms.oms_order_comment", "comment_id");
        tableMap.put("zmn_oms.oms_order_extend", "ext_id");
        tableMap.put("zmn_oms.oms_order_invoice", "invoice_id");
        //tableMap.put("zmn_oms.oms_order_part", "order_id");
        tableMap.put("zmn_oms.oms_order_remark", "order_id");
        tableMap.put("zmn_fico.fico_clearing_master_order", "order_id");
        tableMap.put("zmn_oms.oms_order_comment_extend", "order_id");
        tableMap.put("zmn_oms.oms_order_factory", "fc_order_id");
        tableMap.put("zmn_oms.oms_order_factory_part", "order_id");
        tableMap.put("zmn_oms.oms_order_factory_part_post", "order_id");
        tableMap.put("zmn_oms.oms_order_work_statement", "order_id");
        tableMap.put("zmn_oms.oms_order_nonstandard", "order_id");
        tableMap.put("zmn_oms.oms_order_post", "order_id");
        tableMap.put("zmn_oms.oms_order_attachment", "order_id");
        tableMap.put("zmn_oms.oms_order_tag", "order_id");
        tableMap.put("zmn_oms.oms_order_pay", "order_id");
        tableMap.put("zmn_oms.oms_order_work_visit", "order_id");

    }

    @Autowired
    private ZmnMQSender zmnMQSender;

    @NacosValue(value = "${enable.dts:false}", autoRefreshed = true)
    private boolean enableDts;

    @Override
    public void process(ConsumerRecord<String, byte[]> record) {
        DatumReader<Record> reader = new SpecificDatumReader<>(Record.class);
        Decoder decoder = DecoderFactory.get().binaryDecoder(record.value(), null);
        try {
            Record read = reader.read(null, decoder);
            //将解析的数据进行消费处理
            processConsumer(read);
        } catch (IOException ex) {
            log.error(ex.getMessage(), ex);
        }
    }


    /**
     * 处理消费数据
     *
     * @param record
     */
    private void processConsumer(Record record) {
        if (!enableDts) {
            return;
        } else {
            log.debug("kafka sync enable");
        }
        Operation operation = record.getOperation();
        // 订单这块只过滤 新增 修改 删除的
        boolean supportOperation = Objects.equals(operation, Operation.INSERT) || Objects.equals(operation, Operation.UPDATE) || Objects.equals(operation, Operation.DELETE);
        if (!supportOperation) {
            return;
        }

        // 获取 keyName
        String keyName = getKeyName(record);
        if (StringUtil.isBlank(keyName)) {
            return;
        }

        // 获取 keyValue
        Long keyValue = getKeyValue(record, keyName);
        if (Objects.isNull(keyValue)) {
            return;
        }

        // 根据条件同步订单配件数据
        if (supportPart(record)) {
            sendOrderPartMq(keyValue);
        }

        sendOrderMq(keyValue);
        sendOrderWorkMq(keyValue);

        // 根据条件同步到mos_order
        if (supportMosOrder(record)) {
            sendMosOrder(keyValue);
        }

        // 根据条件同步工单支付凭证
        if (supportOrderPay(record)) {
            sendOrderPayMq(keyValue);
        }

    }

    /**
     * 是否同步到 mos_order
     *
     * @param record
     * @return
     */
    private boolean supportMosOrder(Record record) {
        // 非工单表跳过不同步
        if (!record.getObjectName().toString().startsWith(WORK_CHANGE_TABLE)) {
            return false;
        }

        return true;
    }

    /**
     * 是否同步到 mos_order
     *
     * @param record
     * @return
     */
    private boolean supportOrderPay(Record record) {
        // 非工单支付凭证表跳过不同步
        if (record.getObjectName().toString().startsWith(PAY_CHANGE_TABLE)) {
            return true;
        }

        return false;
    }

    /**
     * 是否支持同步配件
     *
     * @param record
     * @return
     */
    private boolean supportPart(Record record) {

        if (record.getObjectName().toString().startsWith(PART_CHANGE_TABLE)) {
            return true;
        }

        // 非工单表跳过不同步
        if (!record.getObjectName().toString().startsWith(WORK_CHANGE_TABLE)) {
            return false;
        }

        // 算账需要同步
        List<Field> fields = (List<Field>) record.getFields();
        Optional<Field> accountterIdOptional = fields.stream()
                .filter(e -> Objects.equals(e.getName().toString(), "accounter_id"))
                .findFirst();
        if (accountterIdOptional.isPresent()) {

            Field field = accountterIdOptional.get();
            int index = fields.indexOf(field);

            Object beforeImagesObject = record.getBeforeImages();
            Object afterImagesObject = record.getAfterImages();

            // 算账 操作
            if (Objects.nonNull(beforeImagesObject) && Objects.nonNull(afterImagesObject)) {

                List<Object> beforeImages = (List<Object>) beforeImagesObject;
                List<Object> afterImages = (List<Object>) afterImagesObject;
                Integer beforeValue = (Integer) beforeImages.get(index);
                Integer afterValue = (Integer) afterImages.get(index);
                if (NumberUtil.isNullOrZero(Long.valueOf(beforeValue.getValue().toString())) && !NumberUtil.isNullOrZero(Long.valueOf(afterValue.getValue().toString()))) {
                    return true;
                }
            }
        }

        // 配件报销需要同步
        Optional<Field> partReimburseStatusOptional = fields.stream()
                .filter(e -> Objects.equals(e.getName().toString(), "part_reimburse_status"))
                .findFirst();
        if (partReimburseStatusOptional.isPresent()) {

            Field field = partReimburseStatusOptional.get();
            int index = fields.indexOf(field);

            Object beforeImagesObject = record.getBeforeImages();
            Object afterImagesObject = record.getAfterImages();

            // 配件报销 操作
            if (Objects.nonNull(beforeImagesObject) && Objects.nonNull(afterImagesObject)) {

                List<Object> beforeImages = (List<Object>) beforeImagesObject;
                List<Object> afterImages = (List<Object>) afterImagesObject;
                Integer beforeValue = (Integer) beforeImages.get(index);
                Integer afterValue = (Integer) afterImages.get(index);
                if (!NumberUtil.isNullOrZero(Long.valueOf(beforeValue.getValue().toString())) && !NumberUtil.isNullOrZero(Long.valueOf(afterValue.getValue().toString()))) {
                    boolean isDoing = Objects.equals(beforeValue.getValue().toString(), String.valueOf(OrderConsts.PART_REIMBURSE_STATUS_DOING));
                    boolean isSuccess = Objects.equals(afterValue.getValue().toString(), String.valueOf(OrderConsts.PART_REIMBURSE_STATUS_SUCCESS));
                    boolean isFail = Objects.equals(afterValue.getValue().toString(), String.valueOf(OrderConsts.PART_REIMBURSE_STATUS_FAIL));
                    if (isDoing && (isSuccess || isFail)) {
                        return true;
                    }
                }
            }
        }

        return false;
    }

    /**
     * 获取key的值
     *
     * @param record
     * @param keyName
     * @return
     */
    private Long getKeyValue(Record record, String keyName) {

        Long keyValue = null;
        List<Field> fields = (List<Field>) record.getFields();

        Optional<Field> keyFieldOptional = fields.stream()
                .filter(e -> Objects.equals(e.getName().toString(), keyName))
                .findFirst();

        if (!keyFieldOptional.isPresent()) {
            return null;
        }

        Field keyField = keyFieldOptional.get();
        int keyIndex = fields.indexOf(keyField);

        // 新增删除情况 before 和 after 都检查一遍
        Object beforeImagesObject = record.getBeforeImages();
        if (beforeImagesObject != null) {
            List<Object> beforeImages = (List<Object>) beforeImagesObject;
            Integer key = (Integer) beforeImages.get(keyIndex);
            keyValue = Long.valueOf(key.getValue().toString());
        }

        Object afterImagesObject = record.getAfterImages();
        if (Objects.isNull(keyValue) && Objects.nonNull(afterImagesObject)) {
            List<Object> afterImages = (List<Object>) afterImagesObject;
            Integer key = (Integer) afterImages.get(keyIndex);
            keyValue = Long.valueOf(key.getValue().toString());
        }

        return keyValue;
    }

    /**
     * 获取key名称
     *
     * @param record
     * @return
     */
    private String getKeyName(Record record) {

        CharSequence objectName = record.getObjectName();
        if (Objects.isNull(objectName)) {
            return null;
        }

        String tableName = record.getObjectName().toString();
        String key = tableMap.get(tableName);

        if (StringUtils.isEmpty(key)) {
            String simpleTableName = tableName.substring(0, tableName.length() - 2);
            key = tableMap.get(simpleTableName);
        }

        if (StringUtil.isBlank(key)) {
            return null;
        }

        return key;
    }

    /**
     * 工单
     *
     * @param orderId
     */
    private void sendOrderWorkMq(Long orderId) {
        String key = StringUtil.getRandomsString(32);
        log.debug("#oms#dts#orderwork#sendMq key : {} orderId : {}", key, orderId);

        List<Long> orderIds = Arrays.asList(orderId);
        zmnMQSender.send(MqTopicConsts.ZMN_TOPIC_OMS, MqTagConsts.ZMN_TOPIC_OMS_CHANGE_ORDER_WORK, key, JSON.toJSONString(orderIds));
    }

    /**
     * 订单
     *
     * @param orderId
     */
    private void sendOrderMq(Long orderId) {
        String key = StringUtil.getRandomsString(32);
        log.debug("#oms#dts#order#sendMq key : {} orderId : {}", key, orderId);

        zmnMQSender.send(MqTopicConsts.ZMN_TOPIC_OMS, MqTagConsts.ZMN_TOPIC_OMS_CHANGE_ORDER, key, String.valueOf(orderId));
    }

    /**
     * 订单配件
     *
     * @param orderId
     */
    private void sendOrderPartMq(Long orderId) {
        String key = StringUtil.getRandomsString(32);
        log.debug("#oms#dts#orderpart#sendMq key : {} orderId : {}", key, orderId);

        List<Long> orderIds = Arrays.asList(orderId);
        zmnMQSender.send(MqTopicConsts.ZMN_TOPIC_OMS, MqTagConsts.ZMN_TOPIC_OMS_CHANGE_ORDER_PART, key, JSON.toJSONString(orderIds));
    }


    /**
     * 订单评价
     *
     * @param orderId
     */
    private void sendMosOrder(Long orderId) {
        String key = StringUtil.getRandomsString(32);
        if (log.isDebugEnabled()) {
            log.debug("#oms#dts#mos#sendMq key : {} orderId : {}", key, orderId);
        }
        List<Long> orderIdList = Arrays.asList(orderId);
        zmnMQSender.send(MessageQueueTopicConstant.ZMN_TOPIC_MOS, OmsMqTagConsts.MOS_TAG_ORDER_WORK, key, JSON.toJSONString(orderIdList));
    }

    /**
     * 订单支付凭证
     *
     * @param orderId
     */
    private void sendOrderPayMq(Long orderId) {
        String key = StringUtil.getRandomsString(32);
        if (log.isDebugEnabled()) {
            log.debug("#oms#dts#order#pay#sendMq key : {} orderId : {}", key, orderId);
        }
        List<Long> orderIdList = Arrays.asList(orderId);
        zmnMQSender.send(MessageQueueTopicConstant.ZMN_TOPIC_OMS, OmsMqTagConsts.ZMN_TOPIC_OMS_ORDER_PAY_CHANGE_TAG, key, JSON.toJSONString(orderIdList));
    }

}
